package cn.wittyneko.rxbus;


import io.reactivex.functions.Consumer;

/**
 * RxBus默认订阅处理
 * Created by wittytutu on 16-12-22.
 */
public class EventNext implements Consumer<EventBusMsg> {

    private Consumer<? super EventBusMsg> onNext;
    private Consumer<Throwable> onError;

    public EventNext(Consumer<? super EventBusMsg> onNext) {
        this.onNext = onNext;
    }

    public EventNext(Consumer<? super EventBusMsg> onNext, Consumer<Throwable> onError) {
        this.onNext = onNext;
        this.onError = onError;
    }

    @Override
    public void accept(EventBusMsg events) throws Exception {

        // 捕获异常处理
        // RxJava默认的异常处理会执行onError无法再接收消息
        try {
            onNext.accept(events);
        } catch (Throwable e) {
            if (onError != null) {
                //e.fillInStackTrace();
                if (e != null)
                    onError.accept(e);
                else
                    onError.accept(new NullPointerException("RxBus onError"));
            }
        }
    }
}